package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo6MysqlSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义数据源
    val studentDS: DataStream[String] = env.addSource(new MysqlSource)

    studentDS.print()

    env.execute()
  }
}

/**
 * 自定义source 读取mysql中的数据
 */

class MysqlSource extends SourceFunction[String] {
  //用于读取数据
  override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = {
    //1、加载驱动，需要导入mysql的依赖
    Class.forName("com.mysql.jdbc.Driver")
    //2、创建数据库连接
    val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata", "root", "123456")
    //3、编写sql
    val stat: PreparedStatement = con.prepareStatement("select * from students")
    //4、执行查询，获取数据
    val resultSet: ResultSet = stat.executeQuery()
    //5、解析数据
    while (resultSet.next()) {
      //通过字段名获取数据
      val id: String = resultSet.getString("id")
      val name: String = resultSet.getString("name")
      val age: Int = resultSet.getInt("age")
      val sex: String = resultSet.getString("sex")
      val clazz: String = resultSet.getString("clazz")

      //将数据发送到下游
      sourceContext.collect(s"$id,$name,$age,$sex,$clazz")
    }

    //关闭数据量连接
    con.close()
  }

  override def cancel(): Unit = {}
}
